package com.share.hystrix;

import com.netflix.hystrix.HystrixCollapser;
import com.netflix.hystrix.HystrixObservableCollapser;
import com.netflix.hystrix.HystrixObservableCommand;
import rx.functions.Func1;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * author:caifan
 * date:2019/7/15
 */
public class ObservableCollapserGetWordForNumber extends HystrixObservableCollapser<Integer, ObservableCommandNumbersToWords.NumberWord, String, Integer> {
    private final Integer number;
    private final static AtomicInteger counter = new AtomicInteger();

    public ObservableCollapserGetWordForNumber(Integer number) {
        this.number = number;
    }

    public static void resetCmdCounter() {
        counter.set(0);
    }

    public static int getCmdCount() {
        return counter.get();
    }
    @Override
    public Integer getRequestArgument() {
        return number;
    }

    @SuppressWarnings("boxing")
    @Override
    protected HystrixObservableCommand<ObservableCommandNumbersToWords.NumberWord> createCommand(final Collection<HystrixCollapser.CollapsedRequest<String, Integer>> requests) {
        final int count = counter.incrementAndGet();
        System.out.println("Creating batch for " + requests.size() + " requests. Total invocations so far: " + count);

        final List<Integer> numbers = new ArrayList<>();
        for (final HystrixCollapser.CollapsedRequest<String, Integer> request : requests) {
            numbers.add(request.getArgument());
        }

        return new ObservableCommandNumbersToWords(numbers);
    }

    @Override
    protected Func1<ObservableCommandNumbersToWords.NumberWord, Integer> getBatchReturnTypeKeySelector() {
        // Java 8: (final NumberWord nw) -> nw.getNumber();

        return new Func1<ObservableCommandNumbersToWords.NumberWord, Integer>() {
            @Override
            public Integer call(final ObservableCommandNumbersToWords.NumberWord nw) {
                return nw.getNumber();
            }
        };
    }

    @Override
    protected Func1<Integer, Integer> getRequestArgumentKeySelector() {
        // Java 8: return (final Integer no) -> no;

        return new Func1<Integer, Integer>() {
            @Override
            public Integer call(final Integer no) {
                return no;
            }

        };
    }

    @Override
    protected Func1<ObservableCommandNumbersToWords.NumberWord, String> getBatchReturnTypeToResponseTypeMapper() {
        // Java 8: return (final NumberWord nw) -> nw.getWord();

        return new Func1<ObservableCommandNumbersToWords.NumberWord, String>() {
            @Override
            public String call(final ObservableCommandNumbersToWords.NumberWord nw) {
                return nw.getWord();
            }
        };
    }

    @Override
    protected void onMissingResponse(final HystrixCollapser.CollapsedRequest<String, Integer> request) {
        request.setException(new Exception("No word"));
    }

}
